package com.tn.openrpc.jms;

import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.TextMessage;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.springframework.jms.InvalidDestinationException;
import org.springframework.jms.core.JmsTemplate;

/**
 * The endpoint that receives and processes a JMS <code>TextMessage</code>.
 */
public class JmsEndPoint
{
  private static final Logger LOGGER = LoggerFactory.getLogger(JmsEndPoint.class);

  private static final long DEFAULT_TIMEOUT = 1000;

  private static int threadCount = 0;

  private ConnectionFactory connectionFactory;
  private MessageProcessor<TextMessage> messageProcessor;
  private Destination requestDestination;
  private boolean running;
  private long timeOut;

  /**
   * Creates a new <code>JmsEndPoint</code>.
   *
   * @param connectionFactory  the connection factory used to connect to the destinations.
   * @param requestDestination the destination used to send requested.
   * @param messageProcessor   the message processor.
   */
  public JmsEndPoint(ConnectionFactory connectionFactory, Destination requestDestination, MessageProcessor<TextMessage> messageProcessor)
  {
    this(connectionFactory, requestDestination, messageProcessor, DEFAULT_TIMEOUT);
  }

  /**
   * Creates a new <code>JmsEndPoint</code>.
   *
   * @param connectionFactory  the connection factory used to connect to the destinations.
   * @param requestDestination the destination used to send requested.
   * @param messageProcessor   the message processor.
   * @param timeOut            the timeout messages are waited for.
   */
  public JmsEndPoint(
    ConnectionFactory connectionFactory,
    Destination requestDestination,
    MessageProcessor<TextMessage> messageProcessor,
    long timeOut
  )
  {
    this.connectionFactory = connectionFactory;
    this.requestDestination = requestDestination;
    this.messageProcessor = messageProcessor;
    this.timeOut = timeOut;
  }

  /**
   * Returns the running flag.
   */
  public boolean isRunning()
  {
    return running;
  }

  /**
   * Sets the timeout.
   */
  public void setTimeOut(long timeOut)
  {
    this.timeOut = timeOut;
  }

  /**
   * Starts this <code>JmsEndPoint</code>.
   */
  public void start()
  {
    if (!running)
    {
      running = true;

      Thread messageProcessor = new Thread(new MessageMonitor(), "message-processor-" + threadCount++);
      messageProcessor.setDaemon(true);
      messageProcessor.start();
    }
  }

  /**
   * Stops this <code>JmsEndPoint</code>.
   */
  public void stop()
  {
    running = false;
  }

  /**
   * An implementation of <code>Runnable</code> that listens for requests an processes them accordingly.
   */
  private class MessageMonitor implements Runnable
  {
    /**
     * {@inheritDoc}
     */
    @Override
    public void run()
    {
      JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
      jmsTemplate.setReceiveTimeout(timeOut);

      while (running)
      {
        try
        {
          Message message = jmsTemplate.receive(requestDestination);

          if (message == null)
          {
            continue;
          }

          if (!(message instanceof TextMessage))
          {
            LOGGER.error("Unhandled message type encountered: {}", message);
            continue;
          }

          LOGGER.debug("Message received: {}", message);

          messageProcessor.processMessage((TextMessage)message);
        }
        catch (InvalidDestinationException e)
        {
          LOGGER.info("An error occurred sending response message with error message: {}", e.getMessage());
        }
        catch (RuntimeException e)
        {
          LOGGER.error("A runtime error occurred reading the message contents.", e);
        }
        catch (MessageProcessingException e)
        {
          LOGGER.error("An error occurred processing the message.", e);
        }
      }

      threadCount--;
    }
  }
}
